package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.AggregatingState;
import org.apache.flink.api.common.state.AggregatingStateDescriptor;
import org.apache.flink.api.common.state.ReducingState;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/1
 *
 *  计算每种传感器的平均水位
 */
public class Demo6_AggregatingState
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   .keyBy(WaterSensor::getId)
                   .process(new KeyedProcessFunction<String, WaterSensor, String>()
                   {


                       private AggregatingState<Integer, Double> aggregatingState;

                       @Override
                       public void open(Configuration parameters) throws Exception {
                           //初始化状态
                           aggregatingState = getRuntimeContext().getAggregatingState(new AggregatingStateDescriptor<>("sumVC", new AggregateFunction<Integer, MyAcc, Double>()
                           {
                               @Override
                               public MyAcc createAccumulator() {
                                   return new MyAcc(0, 0d);
                               }

                               @Override
                               public MyAcc add(Integer value, MyAcc accumulator) {
                                   accumulator.num += 1;
                                   accumulator.sumVc += value;
                                   return accumulator;
                               }

                               @Override
                               public Double getResult(MyAcc accumulator) {
                                   return accumulator.sumVc / accumulator.num;
                               }

                               @Override
                               public MyAcc merge(MyAcc a, MyAcc b) {
                                   return null;
                               }
                           }, MyAcc.class));
                       }

                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                          //把数据扔进去，自动聚合
                           aggregatingState.add(value.getVc());

                           out.collect(ctx.getCurrentKey() +"平均vc:"+aggregatingState.get());

                       }
                   })
                   .print();
        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyAcc{
        private Integer num;
        private Double sumVc;
    }
}
